package com.etlmaster.rowprocessor.core;

import java.io.IOException;
import java.lang.reflect.Constructor;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternResolver;

import com.etlmaster.executor.annotation.StepInstruction;
import com.etlmaster.executor.trans.core.Line;
import com.etlmaster.executor.trans.core.LineTransProcessor;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.LogWriter;
import com.etlmaster.rowprocessor.disruptor.readEvent.LineEvent;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class Processor extends AbstractOutPutPlugin implements AbstractInputPlugin{

	private String inputStreamId,outPutStreamId;
	private List processorChangeMap;
	private Map paramMap;

	public List processorList=new ArrayList();

	public Processor(String id,String jobId, String name, String code,
			Map<String, Object> paramMap,Map<String, Object> var) {
		super(id,jobId, name, code, paramMap,var);
	}
	public List getProcessorChangeMap() {
		return processorChangeMap;
	}
	public void setProcessorChangeMap(List processorChangeMap) {
		this.processorChangeMap = processorChangeMap;
	}
	protected  void init(){
		for(int i=0;i<processorChangeMap.size();i++){
			Map<String,String> codeMap=(Map<String, String>) processorChangeMap.get(i);

			LineTransProcessor processor = null;
			String changeCode=codeMap.get("changeCode");
			String className = datax_INST_CLASS_MAPP.get(changeCode);
			if (className == null) {
				LogWriter.addLog(this.getJobId(),0,"ERROR","步骤指令{}错误，程序终止"+changeCode);
			} else {
				try{
				Class cls=Class.forName(className);
				Map paramStr= (Map) paramMap.get(changeCode+codeMap.get("index"));
				Constructor<LineTransProcessor> constructor = cls.getConstructor( String.class,String.class,String.class,String.class,String.class,Map.class,Map.class);
				//(fieldIndex, sourceTab,sourceCol,targetTab,targetCol,changeCode, paramStr,var);
				
				processor=constructor.newInstance("","","","",changeCode,paramStr,this.getVariables());
				}catch(Exception e){
					LogWriter.addLog(this.getJobId(),0,"ERROR","生成转换实例时失败,{}",e.getMessage());
				}
			
			}
			processorList.add(processor);
			
		}
	}
	public static Map<String, String> datax_INST_CLASS_MAPP = new HashMap<String, String>();
	static {
		autoScanComponentInstruction();
	}

	public void clearAll(){
		processorList.clear();
	}
	public static void autoScanComponentInstruction() {
		ResourcePatternResolver resourcePatternResolver = new PathMatchingResourcePatternResolver();
		String basePkg = DataCollectionUtils.getProperties("transBasePkg");
		String packageSearchPath = "classpath*:" + basePkg + "**/*.class";
		try {
			Class.forName("org.springframework.core.io.Resource");
			Resource[] resources = resourcePatternResolver
					.getResources(packageSearchPath);
			for (Resource resource : resources) {
				String uri = resource.getURI().toString();
				String className = uri.substring(uri.indexOf(basePkg),
						uri.length() - 6).replace('/', '.');
				Class cls = Class.forName(className);
				if (cls.isAnnotationPresent(StepInstruction.class)) {
					String key = ((StepInstruction) cls
							.getAnnotation(StepInstruction.class)).value()[0];

					if (datax_INST_CLASS_MAPP.containsKey(key)) {
						LogWriter.addLog("WARN","出现重复的指令类{}与{}"
								+ datax_INST_CLASS_MAPP.get(key) + ","
								+ className);
					} else {
						datax_INST_CLASS_MAPP.put(key, className);
					}
				}
			}

		} catch (IOException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}
	}
	public Map getParamMap() {
		return paramMap;
	}
	public void setParamMap(Map paramMap) {
		this.paramMap = paramMap;
	}
	public void process(Line line){
		for(int i=0;i<processorList.size();i++){
			LineTransProcessor processor=(LineTransProcessor) processorList.get(i);
			processor.transField(line);
			if(line.isSkip())
				break;
			LogWriter.addLog(this.getJobId(),0,"debug","processor:{}",line.toString("|"));
		}
		if(!line.isSkip())
			onData(line);
		else
			onErrorData(line);


	}
	protected void onErrorData(Line line){
		 long sequence = JobManager.getJobs().get(this.getJobId()).getErrorringBuffer().next();  // Grab the next sequence
	        try
	        {
	        	LineEvent event = JobManager.getJobs().get(this.getJobId()).getErrorringBuffer().get(sequence); // Get the entry in the Disruptor
	                                                        // for the sequence
	            event.set(line);  // Fill with data
	        }
	        finally
	        {
	        	JobManager.getJobs().get(this.getJobId()).getErrorringBuffer().publish(sequence);
	        }
	}
	protected void onData(Line line)
    {
        long sequence = getRingBuffer().next();  // Grab the next sequence
        try
        {
        	LineEvent event = getRingBuffer().get(sequence); // Get the entry in the Disruptor
                                                        // for the sequence
            event.set(line);  // Fill with data
        }
        finally
        {
            getRingBuffer().publish(sequence);
        }
        this.setCounter(sequence);
        LogWriter.addLog(this.getJobId(),0,"debug","processor sequence:"+sequence);
    }
	public String getInputStreamId() {
		return inputStreamId;
	}
	public void setInputStreamId(String inputStreamId) {
		this.inputStreamId = inputStreamId;
	}
	public String[] getOutPutStreamId() {
		return outPutStreamId.split(",");
	}
	public void setOutPutStreamId(String outPutStreamId) {
		this.outPutStreamId = outPutStreamId;
	}
	@Override
	public void read(Line line) {
		// TODO Auto-generated method stub
		
	}
	@Override
	public void onEvent(LineEvent event, long sequence, boolean endOfBatch)
			throws Exception {
		try{
			process(event.getValue());
		}catch(Exception e){
			onErrorData(event.getValue());
			LogWriter.addLog(getJobId(), 0, "ERROR","line process field !{}",e.getMessage());
			event.getValue().setSkip(true);
		}
	}
	@Override
	public void write() {
		// TODO Auto-generated method stub
		
	}
	@Override
	protected void stop() {
		// TODO Auto-generated method stub
		
	}
}
